//bitmap.h

#ifndef __LSV_BITMAP_BITMAP_H_
#define __LSV_BITMAP_BITMAP_H_

#define CHUNK_SIZE  1048576
#define BITMAP_CHUNK_SIZE 1048576

#define MAX_VOL_SIZE    (uint64_t)1024*1024*1024*1024*256

#define MAX_NUM_BITMAP   (MAX_VOL_SIZE)/LSV_PAGE_SIZE
#define MAX_NUM_CHUNK   (MAX_NUM_BITMAP) * sizeof(struct lsv_bitmap_unit)/BITMAP_CHUNK_SIZE

#define NUM_OF_BITMAP_CHUNK(x)  (x / LSV_PAGE_SIZE * sizeof(struct lsv_bitmap_unit) / BITMAP_CHUNK_SIZE + 1)
#define BITMAP_CHUNK_HEADER_SIZE(x) NUM_OF_BITMAP_CHUNK(x) * sizeof(lsv_chunk_bitmap_t)

#define LSV_BITMAP_SIGN                 "fusionstack lsv"
#define LSV_BITMAP_SNAP_SIGN            "lsv snapshot"
#define LSV_BITMAP_CLONE_SIGN           "lsv clone vol"

#define LSV_BITMAP_USE_HASH             0
#define LSV_BITMAP_HASH_LEN             16

#define private

#define LSV_BITMAP_TRUE                 1
#define LSV_BITMAP_FALSE                0

#define UNLOCATED_TRUNK                 -1

#define LSV_BITMAP_SNAP_TYPE_PERSISTENT 0x00000000
#define LSV_BITMAP_SNAP_TYPE_HERE       0x00000001
#define LSV_BITMAP_SNAP_TYPE_VOLUME     0x00000002
#define LSV_BITMAP_SNAP_TYPE_DELETED    0x00000004

#define LSV_BITMAP_FLAG_PROTECTED       0x00010000
#define LSV_BITMAP_FLAG_DISABLE_GC      0x00000001

#define LSV_BITMAP_CHECK_BITMAP_LBA     0

#define LSV_BITMAP_USE_GLOBAL_LOCK      1
#define LSV_BITMAP_USE_PAGE_LOCK        0
#define LSV_BITMAP_USE_PAGE_LOAD        0
#define LSV_BITMAP_ASYNC_LOAD           0

//#define LSV_BITMAP_ENABLE_HASH_CACHE   
/**
 * 三种状态：
 * - chunk_id == 0
 * - chunk_id有效，且if_ref==0（私有chunk）
 * - chunk_id有效，且if_ref==1（共享chunk，触发COW过程）
 */
typedef struct lsv_chunk_bitmap
{
        uint32_t        chunk_id:31;
        uint32_t        is_ref:1;
        uint32_t        vvol_id:16;
}lsv_chunk_bitmap_t;

//each bitmap represent to a logical page, for example is 4k
typedef struct lsv_bitmap_unit 
{
        uint32_t        chunk_id:32;
        uint32_t        chunk_page_off:16;
        uint32_t        vvol_id:15;
        uint32_t        ref:1;
#if LSV_BITMAP_USE_HASH
        uint8_t         hash[LSV_BITMAP_HASH_LEN];
#endif
#if LSV_BITMAP_CHECK_BITMAP_LBA
        uint64_t        lba;
#endif
}lsv_bitmap_unit_t;
static_assert(sizeof(lsv_bitmap_unit_t) == 8, "lsv_bitmap_unit_t");

typedef struct row1_bitmap_unit 
{
        uint32_t                chunk_id;//:32;
        uint32_t                chunk_page_off:16;
        uint32_t                vvol_id:16;
        uint8_t                 page_bits[256 / 8];
#if LSV_BITMAP_CHECK_BITMAP_LBA
        uint64_t                lba;
        uint64_t                reserved[2]; 
#else
        uint64_t                reserved[3];
#endif
}row1_bitmap_unit_t;

#pragma pack(1)
typedef struct row2_bitmap_page_unit
{
        uint32_t                chunk_id;
        uint16_t                vvol_id;
}row2_bitmap_page_unit_t;
#pragma pack()

#pragma pack(8)

/**
 * 三种状态：
 * - chunk_id == 0
 * - chunk_id 有效，且owner==1 (私有chunk）
 * - chunk_id 有效，且owner==0 (共享chunk）
 *
 * 一个row2_bitmap_unit代表一个chunk，为了数据共享的需要，内部分页进行管理.
 */
typedef struct row2_bitmap_unit //must be 4k alignment.
{
        uint32_t                chunk_id;//:32;
        uint32_t                owner:1;
        row2_bitmap_page_unit_t page_bits[256];
#if LSV_BITMAP_CHECK_BITMAP_LBA
        uint64_t                lba;
        uint64_t                reserved[2]; 
#else
        uint64_t                reserved[3];
#endif
        uint8_t                 __pad[480];
}row2_bitmap_unit_t;

#pragma pack()

static inline void lsv_bitmap_unit_init(lsv_bitmap_unit_t *bunit,
                                        uint32_t chunk_id,
                                        uint32_t page_off,
                                        uint32_t vvol_id) 
{
        bunit->chunk_id = chunk_id; 
        bunit->chunk_page_off = page_off;
        bunit->vvol_id = vvol_id;
}

typedef struct vvol_map_unit
{
        //uint16_t vvol_id;
        //uint8_t  reserved[6];
        uint64_t vol_id;                 //16bytes
}vvol_map_unit_t;

#define MAX_VVOL_COUNT          224             //must considering align.

/**
 * the header is saved in first chunk, we must know first chunk number..
 *
 * vvol_id为本卷卷内id，可以通过vvol_map查到系统的卷id。
 * 之所以增加这层映射，是为了节约存储空间，从8B到4B。
 *
 * 每页数据的bitmap unit，记录了<vvol_id，chunk_id>，page_idx看情况而定（可选）
 * 如果页记录对应的vvol_id==本卷vvol_id，则表示是卷内数据，否则，需要跨卷读取。
 *
 * 只有clone之后，生成了新卷，才涉及跨卷读取。
 *
 * 在一个卷内部，每个快照对应一个lsv_bitmap_header_t结构，构成快照树。
 *
 * chunk_map是指向下层bitmap chunk的变长数组。
 */
typedef struct lsv_bitmap_header
{
        char                    sign[16]; //must be 'fusionstack lsv' for root node.
        uint32_t                version[2];
        uint64_t                root;
        uint32_t                chunk_count;            //32
        uint32_t                active_chunk_id;
        uint32_t                first_child_chunk_id;
        uint32_t                next_chunk_id;
        uint32_t                header_chunk_ids[16];   //48
        uint32_t                type_and_flag;          //102
        uint32_t                vvol_id;
        uint32_t                chunk_map_count;
        uint64_t                create_time;
        uint32_t                reserved2[3];
        uint64_t                node_id;
        char                    name[256];
        uint8_t                 reserved3[108];
        vvol_map_unit_t         vvol_map[MAX_VVOL_COUNT];          //start at 512, first one is current vol.
        lsv_chunk_bitmap_t      chunk_map[0];
}lsv_bitmap_header_t;


#define MAX_COUNT_CACHE         512 //gloconf.bitmap_cache_count
#define PAGE_PER_CHUNK          (BITMAP_CHUNK_SIZE / LSV_PAGE_SIZE)

typedef void (*lsv_bitmap_data_cow_callback_t)(uint8_t * dataptr);

typedef struct lsv_bitmap_cache_unit
{
        uint8_t                 bitmap_cache[BITMAP_CHUNK_SIZE];
        uint32_t                chunk_id;
        uint32_t                vvol_id;
        uint32_t                ratio;
        uint32_t                ref_count;
        uint8_t                 dirty_bits[BITMAP_CHUNK_SIZE / LSV_PAGE_SIZE / 8];
        uint8_t                 valid_bits[BITMAP_CHUNK_SIZE / LSV_PAGE_SIZE / 8];
        uint8_t                 load_bits[BITMAP_CHUNK_SIZE / LSV_PAGE_SIZE / 8];
        lsv_rwlock_t            dirty_lock[BITMAP_CHUNK_SIZE / LSV_PAGE_SIZE];
        lsv_rwlock_t            slot_lock;
}lsv_bitmap_cache_unit_t;

typedef struct lsv_bitmap_cache
{
        lsv_bitmap_cache_unit_t **bitmap_cache;
        uint32_t                cur_pos;
        uint32_t                free_pos;
        lsv_lock_t              lock;
}lsv_bitmap_cache_t;

typedef void (*lsv_bitmap_data_cow_callback_t)(uint8_t * dataptr);
typedef void (*lsv_bitmap_read_snap_callback_t)(int type, lsv_bitmap_unit_t * bitmap, int count);

typedef struct lsv_bitmap_context
{
        void                            *volume_context;
        struct lsv_bitmap_cache         cache_context;
        void *                          hash_cache;
        struct lsv_bitmap_context       *first_child;
        struct lsv_bitmap_context       *next;
        struct lsv_bitmap_context       *parent; 
        struct lsv_bitmap_context       *active; 

        lsv_bitmap_read_snap_callback_t read_snap_cb;
        lsv_bitmap_data_cow_callback_t  data_cow_callback;

        uint8_t                         is_active;
        uint8_t                         is_session;
        uint8_t                         is_loading;
        uint8_t                         is_cloning;
        uint32_t                        header_size;
        uint32_t                        flags;
        lsv_rwlock_t                    rwlock;
        lsv_rwlock_t                    cow_lock;       
        struct lsv_bitmap_header        *bitmap_header;     //persistent data, must be last one.
}lsv_bitmap_context_t;

typedef struct _lsv_bitmap_update 
{
        uint64_t                        lba;
        uint32_t                        new_chunk_id;
        uint32_t                        new_chunk_page_off;        // off/4k
        uint32_t                        old_chunk_id;
        uint32_t                        old_chunk_page_off;
        uint32_t                        repeat_count;
        uint32_t                        snap_id;
} lsv_bitmap_update_t;


#define BITMAP_HEADER_SIZE ((sizeof(struct lsv_bitmap_header) + CHUNK_SIZE) / CHUNK_SIZE) * CHUNK_SIZE
#define BITMAP_HEADER(x) x->bitmap_header
#define BITMAP_CONTEXT(x) ((lsv_bitmap_context_t *)((lsv_volume_proto_t *)x)->bitmap_context)


#define DATAOFF_TO_BITMAPOFF(x) ((x) * sizeof(struct lsv_bitmap_unit) / LSV_PAGE_SIZE)

//initialize bitmap structure.
int lsv_bitmap_init(void *volume_context, uint32_t create_type, uint32_t flag);

//for clone volume.
int lsv_bitmap_init_by_snap(void *parent_volume_context, void *volume_context, const char *snap_name);

int lsv_bitmap_init_by_buf(void *volume_context, uint64_t parent_vol_id, void *buf, uint32_t flags) ;

int lsv_bitmap_is_ready(void *volume_context);

//read one entry
int lsv_bitmap_paged_read(void *volume_context, uint64_t lba, lsv_bitmap_unit_t *bitmap_buf);

//set one entry.
int lsv_bitmap_paged_write(void *volume_context, uint64_t lba, lsv_bitmap_unit_t *bitmap_buf);

int lsv_bitmap_paged_pre_write(void *volume_context, uint64_t off);

int row_bitmap_chunked_read(void *volume_context, uint64_t off, row2_bitmap_unit_t *bitmap_buf);

int row_bitmap_chunked_write(void *volume_context, uint64_t off, row2_bitmap_unit_t *bitmap_buf);

int row_bitmap_chunked_will_cow(void *volume_context, uint64_t off, uint32_t length);
//delete (zero) one entry.
int lsv_bitmap_paged_delete(void *volume_context, uint64_t lba);

//batch read bitmap entries.
int lsv_bitmap_batch_read(void *volume_context, uint64_t lba, uint32_t len, void *bitmap_buf);

int lsv_bitmap_batch_release(void *volume_context, uint64_t off, uint32_t len, void *bitmap_buf);

//batch write bitmap entries.
int lsv_bitmap_batch_write(void *volume_context, uint64_t lba, uint32_t len, void *bitmap_buf);

//free bitmap.
int lsv_bitmap_deinit(void *volume_context);

int lsv_bitmap_lba_check(void *volume_context,uint64_t lba, uint32_t chunk_id, uint32_t chunk_page_off, uint32_t snap_id);

int lsv_bitmap_release_cache(void *volume_context);

int lsv_bitmap_resize(void *volume_context, uint64_t new_size);

int lsv_bitmap_lba_set(void *volume_context, lsv_bitmap_update_t * update_unit);

int lsv_bitmap_vvol_to_vol(void *volume_context, uint32_t vvol_id, uint64_t *vol_id);

int lsv_bitmap_will_cow(void *volume_context, uint64_t off, uint32_t length);

int lsv_bitmap_do_cow(void *volume_context, uint64_t off, uint32_t length);

static inline int lsv_bitmap_paged_lookup(void *volume_context, uint64_t lba, uint64_t * vol_id, uint32_t * chunk_id, uint32_t *chunk_off)
{
        struct lsv_bitmap_unit bu;
        
        int ret = lsv_bitmap_paged_read(volume_context, lba, &bu);
        if(ret)
                return ret;
        
        #if LSV_BITMAP_CHECK_BITMAP_LBA
        assert(bu.lba == 0 || bu.lba == lba); //maybe unlocated
        #endif
        
        *chunk_id = bu.chunk_id;
        *chunk_off = bu.chunk_page_off * LSV_PAGE_SIZE;
        
        lsv_bitmap_vvol_to_vol(volume_context, bu.vvol_id, vol_id);
        
        DINFO("lsv_bitmap_paged_lookup>off:%llu, chunk_id:%u,chunk_page_off:%u\n", (LLU)lba, bu.chunk_id, *chunk_off);
        
        return 0;
}

static inline int lsv_bitmap_paged_set(void *volume_context, uint64_t lba, uint32_t chunk_id, uint32_t chunk_off)
{
        lsv_bitmap_unit_t bu;
        
        bu.chunk_id = chunk_id;
        bu.chunk_page_off = chunk_off / LSV_PAGE_SIZE;
        
        return lsv_bitmap_paged_write(volume_context, lba, &bu);
}

static inline int lsv_bitmap_batch_lookup(void *volume_context, uint64_t lba, uint32_t len, void *bitmap_buf)
{
        return lsv_bitmap_batch_read(volume_context, lba, len, bitmap_buf);
}

static inline void lsv_bitmap_set_read_snap_callback(void *volume_context, lsv_bitmap_read_snap_callback_t callback_ptr)
{
        struct lsv_bitmap_context *bitmap_context = ((lsv_volume_proto_t *)volume_context)->bitmap_context;

        bitmap_context->read_snap_cb = callback_ptr;
}

static inline void lsv_bitmap_set_cow_callback(void *volume_context, lsv_bitmap_data_cow_callback_t callback_ptr)
{
        struct lsv_bitmap_context *bitmap_context = ((lsv_volume_proto_t *)volume_context)->bitmap_context;
        
        bitmap_context->data_cow_callback = callback_ptr;
}

void lsv_bitmap_wlock(void *volume_context);
void lsv_bitmap_rlock(void *volume_context);

void lsv_bitmap_unlock(void *volume_context);

int lsv_bitmap_start_session(void *volume_context);
int lsv_bitmap_commit_dirty(void *volume_context);

#endif /* __LSV_BITMAP_BITMAP_H_ */
